如何应用事件流实现Kafka的消息路由_事件总线(EventBridge)-阿里云帮助中心

本文介绍如何应用事件总线EventBridge的事件流功能实现云消息队列 Kafka 版的消息路由。

前提条件

已开通事件总线EventBridge并授权。
已购买并部署云消息队列 Kafka 版实例，且实例处于服务中状态。具体步骤，请参见购买和部署实例。

背景信息

事件流作为更轻量、实时端到端的流式事件通道，提供轻量流式数据的过滤和转换的能力，在不同的数据仓库之间、数据处理程序之间、数据分析和处理系统之间进行数据同步。源端云消息队列 Kafka 版生产的消息可以通过事件流这个通道被路由到目标端的云消息队列 Kafka 版，无需定义事件总线。更多信息，请参见事件流概述。

支持地域

支持在云消息队列 Kafka 版之间创建事件流的地域有：华东2（上海）、华北2（北京）、华北3（张家口）、中国香港、德国（法兰克福）、新加坡和日本（东京）。

步骤一：在目标端创建事件流

说明

事件流需要在目标端创建，例如如果需要把华北2（北京）的Kafka消息路由到华东2（上海），那么需要在华东2（上海）创建事件流任务。

登录事件总线EventBridge控制台。
在顶部菜单栏，选择地域。
在左侧导航栏，单击事件流。
在事件流页面，单击创建事件流。

在创建事件流面板，设置任务名称和描述，配置以下参数，然后单击保存。

任务创建

在Source（源）配置向导，选择数据提供方为消息队列 Kafka 版，设置以下参数，然后单击下一步。

参数	说明	示例
地域	选择云消息队列 Kafka 版源实例所在的地域。	华北2（北京）
kafka 实例	选择生产云消息队列 Kafka 版消息的源实例。	MQ_INST_115964845466****_ByBeUp3p
Topic	选择生产云消息队列 Kafka 版消息的Topic。	topic
Group ID	选择源实例的消费组名称。请使用独立的消费组来创建事件源，不要和已有的业务混用消费组，以免影响已有的消息收发。	GID_http_1
并发配额（消费者数）	选择源实例的消费者数。	1
消费位点	选择开始消费消息的位点。	最新位点
网络配置	选择路由消息的网络类型。	默认网络
专有网络VPC	选择VPC ID。当网络配置设置为自建公网时需要设置此参数。	vpc-bp17fapfdj0dwzjkd****
交换机	选择vSwitch ID。当网络配置设置为自建公网时需要设置此参数。	vsw-bp1gbjhj53hdjdkg****
安全组	选择安全组。当网络配置设置为自建公网时需要设置此参数。	alikafka_pre-cn-7mz2****
批量推送	批量推送可帮您批量聚合多个事件，当批量推送条数和批量推送间隔（单位：秒）两者条件达到其一时即会触发批量推送。例如：您设置的推送条数为100 条，间隔时间为15 s，在10 s内消息条数已达到100条，那么该次推送则不会等15 s后再推送。	开启
批量推送条数	调用函数发送的最大批量消息条数，当积压的消息数量到达设定值时才会发送请求，取值范围为 [1,10000]。	100
批量推送间隔（单位：秒）	调用函数的间隔时间，系统每到间隔时间点会将消息聚合后发给函数计算，取值范围为[0,15]，单位为秒。0秒表示无等待时间，直接投递。	3

在Filtering（过滤）、Transform（转换）配置向导，设置事件过滤、转换规则，单击下一步。事件转换的配置说明，请参见使用函数计算实现消息数据清洗。

在Sink（目标）配置向导，选择服务类型为消息队列 Kafka 版，配置以下参数，单击保存。

参数	说明	示例
实例ID	选择已创建的云消息队列 Kafka 版实例。	test
Topic	选择已创建实例中的Topic。	test
确认模式（ack）	选择云消息队列 Kafka 版接收到数据后给客户端发出的确认信号。	None
消息体（Value）	事件总线EventBridge通过JSONPath提取事件中的数据，将指定的事件内容路由到事件目标。	部分事件 `$.data.value`
消息键值（Key）	事件总线EventBridge通过JSONPath提取事件中的数据，将指定的事件内容路由到事件目标。	部分事件 `$.data.key`

任务属性
设置事件流的重试策略及死信队列。更多信息，请参见重试和死信。

返回事件流页面，找到创建好的事件流，在其右侧操作栏，单击启用。
在提示对话框，阅读提示信息，然后单击确认。
启用事件流后，会有30秒~60秒的延迟时间，您可以在事件流页面的状态栏查看启动进度。

步骤二：测试验证

登录云消息队列 Kafka 版控制台。
在顶部菜单栏，选择步骤一：在目标端创建事件流中源实例所在的地域。
在左侧导航栏，单击实例列表。
在实例列表页面，找到步骤一：在目标端创建事件流中配置的源实例，在其操作列，单击详情。
在左侧导航栏，单击Topic 管理。
在Topic列表，单击步骤一：在目标端创建事件流中配置的源实例的Topic名称。
在Topic详情页面，单击右上角的体验发送消息。
在快速体验消息收发面板，选择发送方式为控制台，然后配置消息 Key和消息内容，单击确定。
在源实例完成生产消息后，返回实例列表页面。
在实例列表页面，找到步骤一：在目标端创建事件流中配置的目标实例，在其操作列，单击详情。
在左侧导航栏，单击Topic 管理。
在Topic列表，单击步骤一：在目标端创建事件流中配置的目标实例的Topic名称。
在Topic详情页面，单击消息查询。
配置查询方式、分区和时间点，单击查询。
查看查询到的Key和Value值是否与生产的消息一致。